package com.mytest.provider.ff.listen;

import com.alibaba.fastjson.JSONObject;
import com.mytest.provider.ff.controller.ffController;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;


@Component
public class MqListen {

    @Autowired
    ffController ffController;

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "nebula.create.page.queue", durable = "true"),
            exchange = @Exchange(
                    value = "mytest.item.exchange",
                    ignoreDeclarationExceptions = "true",
                    type = ExchangeTypes.TOPIC,
                    durable = "ture"),
            key = "mytest.select"))
    public void listenCreateA(String news, Message message, Channel channel) throws Exception {
        System.out.println("1111111111111111111");
        if (news == null) {
            return;
        }
        // 获取消息
        try {
            System.out.println("接受到的消息A是:"+news);

            ffController.mq(news);

            /*Map<String,Object> arg = new HashMap<String, Object>();
            arg.put("x-message-ttl",6000);
            //创建一个交换机
            channel.queueDeclare("normalQueue",true,false,false,arg);

            AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
            //持久化消息
            builder.deliveryMode(2);
            //设置过期时间
            builder.expiration("6000");
            AMQP.BasicProperties properties = builder.build();
            //发布一条消息
            channel.basicPublish("exchange","routingkey",true,properties,"hello".getBytes());*/

            //手动确认提交， 消息的索引，是否手动确认
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
            System.out.println("消费消息确认" + message.getMessageProperties().getConsumerQueue() + "，接收到了回调方法");
        } catch (IOException e) {
            //拒绝一条消息
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            e.printStackTrace();
        }
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "mytest.create.page.queue", durable = "true"),
            exchange = @Exchange(
                    value = "mytest.item.exchange",
                    ignoreDeclarationExceptions = "true",
                    type = ExchangeTypes.TOPIC),
            key = "mmm.aaa"))
    public void listenCreateB(String news, Message message, Channel channel) throws Exception {
        System.out.println("1111111111111111112");
        if (news == null) {
            return;
        }
        // 获取消息
        try {
            System.out.println("接受到的消息B是:"+news);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
            System.out.println("消费消息确认" + message.getMessageProperties().getConsumerQueue() + "，接收到了回调方法");
        } catch (IOException e) {
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            e.printStackTrace();
        }
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "my.create.page.queue", durable = "true"),
            exchange = @Exchange(
                    value = "mytest.item.exchange",
                    ignoreDeclarationExceptions = "true",
                    type = ExchangeTypes.TOPIC),
            key = "mmm.zzz"))
    public void listenCreateC(String news, Message message, Channel channel) throws Exception {
        System.out.println("1111111111111111113");
        if (news == null) {
            return;
        }
        // 获取消息
        try {
            System.out.println("接受到的消息C是:"+news);
            //消息确认
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);

            //队列的解绑
            channel.queueUnbind("mytest.create.page.queue", "mytest.item.exchange","mmm.kkk");
            System.out.println("消费消息确认" + message.getMessageProperties().getConsumerQueue() + "，接收到了回调方法");
        } catch (IOException e) {
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            e.printStackTrace();
        }
    }


    @RabbitListener(queues = "fanout_email_queue")
    public void process(Message message, @Headers Map<String, Object> headers, Channel channel) throws Exception {
        // 获取消息Id
        String messageId = message.getMessageProperties().getMessageId();
        String msg = new String(message.getBody(), "UTF-8");

        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        System.out.println("消息接收时间:"+sdf.format(new Date()));
        System.out.println("邮件消费者获取生产者消息msg:"+msg+",消息id"+messageId);

        JSONObject jsonObject = JSONObject.parseObject(msg);
        Integer timestamp = jsonObject.getInteger("timestamp");

        try {
            /*int result  = 1/timestamp;
            System.out.println("result"+result);*/
            // // 手动ack
            Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
            // 手动签收,false 不批量签收 ，true 批量签收
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            //拒绝消费消息（丢失消息） 给死信队列
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        }

        System.out.println("执行结束....");
    }

}


